package cn.zwx.common.canal.listener;

import cn.zwx.common.canal.bo.CanalEntryBO;
import cn.zwx.common.canal.event.AbstractCanalEvent;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.util.CollectionUtils;

import java.util.List;


/**
 * canal监听基类
 * 只监听我们自己封装的事件
 * @author zhangwenxue
 */
public abstract class AbstractCanalListener<EVENT extends AbstractCanalEvent> implements ApplicationListener<EVENT> {

    private final Logger logger = LoggerFactory.getLogger(AbstractCanalListener.class);

    @Override
    public void onApplicationEvent(EVENT event) {
        // 获取监听对象
        CanalEntryBO entryBO = event.getEntry();
        CanalEntry.Entry entry = entryBO.getEntry();
        // 获取数据库名称
        String database = entry.getHeader().getSchemaName();
        // 获取表名称
        String tableName = entry.getHeader().getTableName();
        // 获取变动元素
        CanalEntry.RowChange rowChange;
        try {
            rowChange =  CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        }catch (InvalidProtocolBufferException e){
            logger.error("canalEntry_parser_error,根据CanalEntry获取RowChange失败！", e);
            return;
        }

        // 具体处理交子类
        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
        if (CollectionUtils.isEmpty(rowDataList)){
            logger.info("未获取到变动数据的rowData");
            return;
        }
        rowDataList.forEach((rowData) ->
            this.doChange(database,tableName,rowData,entryBO.getCanalInstanceName(),entryBO.getInstanceTopic())
        );

    }

    /**
     * 对应具体业务处理逻辑
     * @param database 数据库名称
     * @param tableName 表名称
     * @param rowData  变动数据
     * @param instanceName 实例名称
     * @param topic 消息的Topic
     * @author zhangwenxue
     **/
    protected abstract void doChange(String database, String tableName, CanalEntry.RowData rowData,String instanceName,String topic);




}

